//! RESP协议解析层演示
//! 
//! 展示如何使用RESP->command解析层处理Redis客户端连接

use std::sync::Arc;
use std::thread;
use std::time::Duration;
use std::io::{self, Read, Write};

use redis_rs2::{
    protocol::{CommandProcessor, CommandProcessorConfig, NetworkStream, TcpStreamAdapter},
    storage::{SingleThreadCommandDispatcher, Storage},
    event::ResponseWritebackDriver,
};

/// 演示用的模拟网络流
struct DemoNetworkStream {
    /// 模拟接收的数据
    data: Vec<u8>,
    /// 读取位置
    pos: usize,
    /// 写入的响应数据
    responses: Vec<u8>,
    /// 是否连接
    connected: bool,
    /// 模拟的远程地址
    remote_addr: String,
}

impl DemoNetworkStream {
    /// 创建演示流，包含多个Redis命令
    fn with_redis_commands() -> Self {
        // 模拟客户端发送的Redis命令（pipeline）
        let mut data = Vec::new();
        
        // PING 命令
        data.extend_from_slice(b"*1\r\n$4\r\nPING\r\n");
        
        // SET key1 value1 命令
        data.extend_from_slice(b"*3\r\n$3\r\nSET\r\n$4\r\nkey1\r\n$6\r\nvalue1\r\n");
        
        // GET key1 命令
        data.extend_from_slice(b"*2\r\n$3\r\nGET\r\n$4\r\nkey1\r\n");
        
        // SET key2 value2 命令
        data.extend_from_slice(b"*3\r\n$3\r\nSET\r\n$4\r\nkey2\r\n$6\r\nvalue2\r\n");
        
        // EXISTS key1 key2 key3 命令
        data.extend_from_slice(b"*4\r\n$6\r\nEXISTS\r\n$4\r\nkey1\r\n$4\r\nkey2\r\n$4\r\nkey3\r\n");
        
        Self {
            data,
            pos: 0,
            responses: Vec::new(),
            connected: true,
            remote_addr: "127.0.0.1:54321".to_string(),
        }
    }
    
    /// 获取写入的响应数据
    pub fn get_responses(&self) -> &[u8] {
        &self.responses
    }
    
    /// 将响应转换为字符串（用于显示）
    pub fn responses_as_string(&self) -> String {
        String::from_utf8_lossy(&self.responses).to_string()
    }
}

impl NetworkStream for DemoNetworkStream {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if self.pos >= self.data.len() {
            // 模拟连接关闭
            return Ok(0);
        }
        
        let remaining = self.data.len() - self.pos;
        let to_copy = std::cmp::min(buf.len(), remaining);
        
        buf[..to_copy].copy_from_slice(&self.data[self.pos..self.pos + to_copy]);
        self.pos += to_copy;
        
        // 模拟网络延迟
        thread::sleep(Duration::from_millis(1));
        
        Ok(to_copy)
    }
    
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.responses.extend_from_slice(buf);
        Ok(buf.len())
    }
    
    fn flush(&mut self) -> io::Result<()> {
        // 在真实实现中会刷新缓冲区
        Ok(())
    }
    
    fn is_connected(&self) -> bool {
        self.connected && self.pos < self.data.len()
    }
    
    fn remote_addr(&self) -> Option<String> {
        Some(self.remote_addr.clone())
    }
    
    fn shutdown(&mut self) -> io::Result<()> {
        self.connected = false;
        Ok(())
    }
}

fn main() {
    println!("🚀 Redis-rs2 RESP协议解析层演示");
    println!("{}", "=".repeat(60));
    
    // 1. 创建存储层组件
    println!("\n📦 1. 创建存储层组件");
    let mut storage_dispatcher = SingleThreadCommandDispatcher::new();
    println!("  ✅ 存储命令分发器已创建");
    
    // 2. 创建响应回写驱动器
    println!("\n📡 2. 创建响应回写驱动器");
    let response_driver = ResponseWritebackDriver::new("demo-response-driver", 5000);
    println!("  ✅ 响应回写驱动器已创建");
    
    // 3. 创建命令处理器配置
    println!("\n⚙️ 3. 配置命令处理器");
    let config = CommandProcessorConfig {
        read_buffer_size: 1024,
        connection_timeout: Duration::from_secs(30),
        cleanup_interval: Duration::from_secs(10),
        max_pipeline_commands: 50,
        command_timeout: Duration::from_secs(5),
    };
    println!("  ✅ 命令处理器配置完成");
    
    // 4. 创建命令处理器
    println!("\n🎯 4. 创建RESP协议命令处理器");
    let processor = CommandProcessor::new(storage_dispatcher, response_driver, Some(config));
    println!("  ✅ 命令处理器已创建");
    
    // 5. 启动命令处理器
    println!("\n🚀 5. 启动命令处理器");
    if let Err(e) = processor.start() {
        eprintln!("❌ 启动失败: {}", e);
        return;
    }
    println!("  ✅ 命令处理器启动成功");
    
    // 6. 创建模拟客户端连接
    println!("\n🌐 6. 模拟客户端连接");
    let demo_stream = Box::new(DemoNetworkStream::with_redis_commands());
    println!("  📨 模拟命令流包含:");
    println!("    - PING");
    println!("    - SET key1 value1");
    println!("    - GET key1");
    println!("    - SET key2 value2");
    println!("    - EXISTS key1 key2 key3");
    
    // 7. 处理连接
    println!("\n🔄 7. 处理客户端连接");
    match processor.handle_connection(demo_stream) {
        Ok(()) => {
            println!("  ✅ 连接处理已启动");
        }
        Err(e) => {
            eprintln!("  ❌ 连接处理失败: {}", e);
            return;
        }
    }
    
    // 8. 等待命令处理完成
    println!("\n⏳ 8. 等待命令处理完成");
    for i in 1..=10 {
        thread::sleep(Duration::from_secs(1));
        
        let stats = processor.get_stats();
        println!("  📊 第{}秒 - 已处理命令: {} 成功: {} 失败: {} 解析错误: {}", 
            i, stats.total_commands_processed, stats.successful_commands, 
            stats.failed_commands, stats.parse_errors);
        
        if stats.active_connections == 0 {
            println!("  ✅ 所有连接已处理完成");
            break;
        }
    }
    
    // 9. 查看最终统计
    println!("\n📊 9. 最终统计信息");
    processor.print_stats();
    
    // 10. 演示直接使用RespParser
    println!("\n🔧 10. 直接使用RespParser演示");
    demonstrate_resp_parser();
    
    // 11. 演示pipeline命令处理
    println!("\n🔗 11. Pipeline命令处理演示");
    demonstrate_pipeline_processing();
    
    // 12. 停止命令处理器
    println!("\n🛑 12. 停止命令处理器");
    if let Err(e) = processor.stop() {
        eprintln!("❌ 停止失败: {}", e);
    } else {
        println!("  ✅ 命令处理器已安全停止");
    }
    
    println!("\n🎉 RESP协议解析层演示完成！");
    print_summary();
}

/// 演示直接使用RespParser
fn demonstrate_resp_parser() {
    use redis_rs2::protocol::RespParser;
    
    let mut parser = RespParser::new();
    
    // 解析各种RESP类型
    let test_cases = vec![
        ("+OK\r\n", "简单字符串"),
        ("$5\r\nhello\r\n", "批量字符串"), 
        (":42\r\n", "整数"),
        ("-ERR unknown command\r\n", "错误"),
        ("*2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n", "数组"),
        ("$-1\r\n", "空值"),
    ];
    
    for (resp_data, desc) in test_cases {
        parser.reset(); // 重置解析器
        parser.feed_data(resp_data.as_bytes());
        
        match parser.parse_value() {
            Ok(Some(value)) => {
                println!("  ✅ {}: {:?}", desc, value);
            }
            Ok(None) => {
                println!("  ⚠️ {}: 数据不完整", desc);
            }
            Err(e) => {
                println!("  ❌ {}: 解析错误 - {:?}", desc, e);
            }
        }
    }
}

/// 演示pipeline命令处理
fn demonstrate_pipeline_processing() {
    use redis_rs2::protocol::RespParser;
    
    let mut parser = RespParser::new();
    
    // 模拟pipeline: 连续的多个命令
    let pipeline_data = b"*2\r\n$4\r\nPING\r\n*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n*2\r\n$3\r\nGET\r\n$3\r\nkey\r\n";
    
    println!("  📦 Pipeline数据: {} 字节", pipeline_data.len());
    parser.feed_data(pipeline_data);
    
    let mut command_count = 0;
    while let Ok(Some(value)) = parser.parse_value() {
        command_count += 1;
        match parser.resp_to_command(value) {
            Ok(command) => {
                println!("  🔧 命令 {}: {} - 参数: {:?}", 
                    command_count, command.name(), command.to_args());
            }
            Err(e) => {
                println!("  ❌ 命令 {} 转换失败: {:?}", command_count, e);
            }
        }
    }
    
    println!("  ✅ 成功解析 {} 个pipeline命令", command_count);
}

/// 打印总结信息
fn print_summary() {
    println!("\n💡 总结:");
    println!("  ✅ RESP协议解析器工作正常");
    println!("  ✅ 支持所有标准RESP数据类型");
    println!("  ✅ 命令转换和验证功能正常");
    println!("  ✅ Pipeline命令处理功能正常");
    println!("  ✅ 连接管理和生命周期控制正常");
    println!("  ✅ 与存储层和响应回写层集成正常");
    println!("  ✅ 统计监控和错误处理完整");
    
    println!("\n🔧 关键特性:");
    println!("  📡 支持多连接并发处理");
    println!("  🔗 支持Redis pipeline命令");
    println!("  🚀 与现有事件驱动架构集成");
    println!("  📊 完整的监控和统计功能");
    println!("  🛡️ 健壮的错误处理机制");
    println!("  🔄 优雅的连接生命周期管理");
    
    println!("\n🌟 RESP协议解析层已准备好为网络层提供服务！");
}